package com.yanggu.flink.datastream_api.multi_stream_transform.combine_stream

import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, WatermarkStrategy}
import org.apache.flink.api.common.state.ValueStateDescriptor
import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

/**
 * 实时对账的需求。
 * app的支付事件和第三方支付事件将互相等待5s，如果等不到对应的支付事件，需要输出报警信息
 */
object BillCheckExample {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    //来自app的支付日志
    val appStream = env
      .fromElements(
        ("order-1", "app", 1000L),
        ("order-2", "app", 2000L)
      )
      .assignTimestampsAndWatermarks(WatermarkStrategy
        .forMonotonousTimestamps()
        .withTimestampAssigner(new SerializableTimestampAssigner[(String, String, Long)] {
          override def extractTimestamp(element: (String, String, Long), recordTimestamp: Long) = element._3
        })
      )

    //来自第三方支付平台的支付日志
    val thirdPartStream = env
      .fromElements(
        ("order-1", "third-party", "success", 3000L),
        ("order-3", "third-party", "success", 4000L)
      )
      .assignTimestampsAndWatermarks(WatermarkStrategy
        .forMonotonousTimestamps()
        .withTimestampAssigner(new SerializableTimestampAssigner[(String, String, String, Long)] {
          override def extractTimestamp(element: (String, String, String, Long), recordTimestamp: Long) = element._4
        })
      )

    appStream
      .connect(thirdPartStream)
      .keyBy(data => data._1, data => data._1)
      .process(new OrderMatchResult)
      .print("对账结果>>>>>")

    env.execute()

  }


}

class OrderMatchResult extends KeyedCoProcessFunction[String, (String, String, Long), (String, String, String, Long), String] {

  private lazy val appValueState = getRuntimeContext.getState(
    new ValueStateDescriptor[(String, String, Long)]("appValueState", classOf[(String, String, Long)]))

  private lazy val thirdValueState = getRuntimeContext.getState(
    new ValueStateDescriptor[(String, String, String, Long)]("thirdValueState", classOf[(String, String, String, Long)])
  )

  //app数据
  override def processElement1(value: (String, String, Long),
                               ctx: KeyedCoProcessFunction[String, (String, String, Long), (String, String, String, Long), String]#Context,
                               out: Collector[String]): Unit = {
    //如果三方有数据
    if (thirdValueState.value() != null) {
      out.collect(s"三方对账成功, app数据: $value, 三方数据: ${thirdValueState.value()}")
      thirdValueState.clear()
    } else {
      //更新状态
      appValueState.update(value)
      //注册定时器, 等待另外一条流的数据
      ctx.timerService().registerEventTimeTimer(value._3 + 5000L)
    }
  }

  //三方支付数据
  override def processElement2(value: (String, String, String, Long),
                               ctx: KeyedCoProcessFunction[String, (String, String, Long), (String, String, String, Long), String]#Context,
                               out: Collector[String]): Unit = {
    //如果app中有数据
    if (appValueState.value() != null) {
      out.collect(s"对账成功, app数据: ${appValueState.value()}, 三方数据: $value")
      appValueState.clear()
    } else {
      //更新状态
      thirdValueState.update(value)
      //注册定时器, 等待另外一条流的数据
      ctx.timerService().registerEventTimeTimer(value._4 + 5000L)
    }
  }

  override def onTimer(timestamp: Long,
                       ctx: KeyedCoProcessFunction[String, (String, String, Long), (String, String, String, Long), String]#OnTimerContext,
                       out: Collector[String]): Unit = {
    //定时器触发，判断状态，如果某个状态不为空，说明另一条流中事件没来
    if (appValueState.value() != null) {
      out.collect(s"对账失败, app数据: ${appValueState.value()}, 三方平台数据未找到")
      appValueState.clear()
    }
    if (thirdValueState.value() != null) {
      out.collect(s"对账失败, 三方平台数据: ${thirdValueState.value()}, app数据未找到")
      thirdValueState.clear()
    }

  }

}
